/*
 * JGroupsGroupConnection.java    0.0.1    21/dic/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transparency.impl;

import java.io.Serializable;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.jgroups.Channel;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelException;
import org.jgroups.ChannelNotConnectedException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.mentalsmash.tazio.commons.TazioRuntimeException;
import org.mentalsmash.tazio.commons.identifiers.NID;
import org.mentalsmash.tazio.transparency.commons.messages.TazioMessage;
import org.mentalsmash.tazio.transparency.interfaces.Dispatcher;
import org.mentalsmash.tazio.transparency.interfaces.GroupConnection;
import org.mentalsmash.tazio.transparency.interfaces.GroupState;
import org.mentalsmash.tazio.transparency.interfaces.MessageFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.internal.Preconditions;
import com.google.inject.name.Named;

/**
 * This is the main {@link GroupConnection} implementation, which exploits the
 * JGroups communication framework in order to send message to the Replica
 * Managers connected together in the cluster.
 * 
 * @version 0.0.1 21/dic/2009
 * @author Enrico Savioli <savioli.enrico@gmail.com>
 */
public class JGroupsGroupConnection extends ReceiverAdapter implements
	GroupConnection {

    private final static Logger log = LoggerFactory
	    .getLogger(JGroupsGroupConnection.class);

    private Lock lock = new ReentrantLock();

    private final String CLUSTERNAME;
    private final String CONFFILE;
    private Channel channel;
    private final MessageFactory messageFactory;
    private final Dispatcher dispatcher;

    @Inject
    public JGroupsGroupConnection(MessageFactory messageFactory,
	    Dispatcher dispatcher, @Named("clustername") String CLUSTERNAME,
	    @Named("jgroupsConfFile") String CONFFILE) {

	this.CLUSTERNAME = CLUSTERNAME;
	this.CONFFILE = CONFFILE;
	this.messageFactory = messageFactory;
	this.dispatcher = dispatcher;

	try {
	    channel = new JChannel(this.CONFFILE);
	} catch (ChannelException e) {
	    String msg = "Error occurred during the creation of the JGroups channel";
	    log.error(msg);
	    e.printStackTrace();
	}

	try {
	    connect();
	} catch (ChannelException e) {
	    log.error("Cannot connect to JGroups!", e);
	    throw new TazioRuntimeException(e);
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.GroupConnection#connect()
     */
    @Override
    public void connect() throws ChannelException {

	channel.setReceiver(this);
	channel.connect(this.CLUSTERNAME);

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.GroupConnection#disconnect()
     */
    @Override
    public void disconnect() {

	channel.close();

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.GroupConnection#getClusterName()
     */
    @Override
    public String getClusterName() {

	return this.CLUSTERNAME;

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.GroupConnection#getGroupState()
     */
    @Override
    public GroupState getGroupState() {

	throw new RuntimeException("Not implemented yet");

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.GroupConnection#getLessLoadedMember()
     */
    @Override
    public NID getLessLoadedMember() {

	throw new RuntimeException("Not implemented yet");

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.GroupConnection#getNumberOfConnections
     * (org.mentalsmash.tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public int getNumberOfConnections(NID id) {

	throw new RuntimeException("Not implemented yet");

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.GroupConnection#isConnected()
     */
    @Override
    public boolean isConnected() {

	return channel.isConnected();

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.GroupConnection#send(java.io.Serializable
     * )
     */
    @Override
    public <T extends Serializable> void send(T message) {

	Preconditions.checkNotNull(message);
	TazioMessage tazioMessage = messageFactory.encapsulateMessage(message);
	Message toSend = new Message(null, null, tazioMessage);

	try {
	    log.debug("Sending message : "+message);
	    channel.send(toSend);
	} catch (ChannelNotConnectedException e) {
	    log.error("Not connected to the group communication subsystem.", e);
	    throw new TazioRuntimeException(e);
	} catch (ChannelClosedException e) {
	    log.error("The JGroups Channel is closed.", e);
	    throw new TazioRuntimeException(e);
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transparency.GroupConnection#updateState(org.
     * mentalsmash.tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public void updateState(NID id) {

	throw new RuntimeException("Not implemented yet");

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.jgroups.ReceiverAdapter#receive(org.jgroups.Message)
     */
    @Override
    public void receive(Message msg) {
	// TODO Auto-generated method stub

	try {
	    lock.lock();

	    log.debug("New message received");
	    System.gc();
	    
	    TazioMessage tMessage = (TazioMessage) msg.getObject();
	    
	    log.debug("Sender: " + tMessage.getMsgID().getNodeID());

	    dispatcher.handleMessage(tMessage);

	} finally {
	    lock.unlock();
	}

    }

}
